package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.hmdp.dto.Result;
import com.hmdp.dto.UserDTO;
import com.hmdp.entity.SeckillVoucher;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.RedisIDWorker;
import com.hmdp.utils.SimpleRedisLock;
import com.hmdp.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopContext;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Service
@Slf4j
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    @Resource
    private ISeckillVoucherService seckillVoucherService;
    @Resource
    private RedisIDWorker redisIDWorker;
    @Resource
    private RedissonClient redissonClient;
    @Resource
    private StringRedisTemplate stringRedisTemplate;

    // Stream消息队列名
    private static final String QueueName = "stream.orders";

    // DefaultRedisScript是一个实现类，泛型是返回值类型Long
    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;

    // 初始化lua脚本\设置
    static {
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        // 设定脚本位置,new ClassPathResource会自动取resources下面找文件
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
        // 设定返回值类型
        SECKILL_SCRIPT.setResultType(Long.class);
    }

    // 线程池
    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

    // 线程任务初始化执行
    @PostConstruct // 这个类初始化就执行
    private void init() {
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
    }


    /**
     * VoucherOrderHandler
     *
     * @param:
     * @return:
     * @description: Stream消息队列线程任务
     * <p>
     * [   seckillVoucher Redis中库存和重复订单判断--库存扣减--订单set增加用户id\发送消息 (lua脚本) =>
     * [=> VoucherOrderHandler 线程任务:取出消息 => hanldeVoucherAndACK 解析消息(获取id)\ACK
     *      => handlerVoucherOrder 获取分布式锁 => createVoucherOrderBlock 乐观锁,扣减库存,创建订单(数据库操作)
     * [=> handlePendingList 线程任务:处理异常消息-ACK
     **/
    private class VoucherOrderHandler implements Runnable {

        @Override
        public void run() {
            while (true) {
                try {
                    // 从stream消息队列获取消息, XREADGROUP GROUP g1 c1 BLOCK 2000 STREAMS stream.orders >
                    // BLOCK 2000，取不到会阻塞2秒，因为不会很频繁
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create(QueueName, ReadOffset.lastConsumed()));
                    // 失败,说明没有消息,continue
                    if (list == null || list.isEmpty()) {
                        continue;
                    }
                    // 成功,可以下单
                    // 解析消息，<String, Object, Object>-- 消息id  消息键值对
                    hanldeVoucherAndACK(list);
                } catch (Exception e) {
                    // 处理发生异常的情况, 即pending_list中存储的异常消息（没有ack）
                    log.error("###### 消息处理异常,handlePendingList开始处理pending_list中存储的异常消息");
                    handlePendingList();
                }
            }
        }
    }

    /**
     * @param: []
     * @return: void
     * @description: 处理异常消息队列 pending_list
     **/
    private void handlePendingList() {
        while (true) {
            try {
                // 从stream消息队列获取消息, XREADGROUP GROUP g1 c1 STREAMS stream.orders 0
                // 0表示读的是pendingList
                List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1),
                        StreamOffset.create(QueueName, ReadOffset.from("0")));
                // 失败,说明pendingList没有异常消息, break
                if (list == null || list.isEmpty()) {
                    break;
                }
                // 成功, 可以处理异常消息
                hanldeVoucherAndACK(list);
            } catch (Exception e) {
                // 处理发生异常的情况, 即pending_list
                // 这里没必要写递归调用自己, 因为本身就在while循环里
                log.error("###### pendingList消息处理异常");
                try {
                    // 暂停一下，防止太频繁
                    Thread.sleep(50);
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
            }
        }
        log.info("###### pendingList消息处理完毕！");
    }

    /**
     * @param: [list]
     * @return: void
     * @description: 解析查询得到的消息, 创建订单, ACK消息
     **/
    private void hanldeVoucherAndACK(List<MapRecord<String, Object, Object>> list) {
        // 解析消息，<String, Object, Object>-- 消息id  消息键值对
        MapRecord<String, Object, Object> mapRecord = list.get(0);
        Map<Object, Object> values = mapRecord.getValue();
        VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
        // 创建订单
        handlerVoucherOrder(voucherOrder);
        // ACk确认, SACK stream.orders g1 id
        stringRedisTemplate.opsForStream().acknowledge(QueueName, "g1", mapRecord.getId());
    }

    /**
     * VoucherOrderHandler
     *
     * @param:
     * @return:
     * @description: 阻塞队列取消息, 处理消息(过时方案)
     **/
    /* // 阻塞队列,某个线程尝试从此队列获取元素,若队列中没有元素 该线程就会被阻塞
    // 需要限制阻塞队列长度,因为使用的是jvm的内存
    private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);
    // 线程任务
    private class VoucherOrderHandler implements Runnable {
        @Override
        public void run() {
            while (true) {
                try {
                    // 从阻塞队列获取订单信息
                    VoucherOrder voucherOrder = orderTasks.take();
                    // 创建订单
                    handlerVoucherOrder(voucherOrder);
                } catch (InterruptedException e) {
                    log.error("###### 处理订单信息异常,线程任务。" + e);
                }
            }
        }
    } */

    // 上锁后操作数据库，感觉其实可以整合到上一级函数
    private void handlerVoucherOrder(VoucherOrder voucherOrder) {
        Long userId = voucherOrder.getUserId();
        // Redisson分布式锁，这里这个锁其实没太大必要，因为放入阻塞队列之前就已经做过判断
        RLock lock = redissonClient.getLock("order:" + userId);
        boolean flag = lock.tryLock();
        if (!flag) {
            // 获取锁失败，错误或者重试
            log.error("同一用户不允许重复下单");
            return;
        }
        try {
            // 获取锁之后创建订单
            proxy.createVoucherOrderBlock(voucherOrder);
        } finally {
            // 释放锁
            lock.unlock();
        }
    }

    /**
     * @param: [voucherId, userId]
     * @return: com.hmdp.dto.Result
     * @description: 扣减库存（乐观锁），insert订单（订单实体类不用创建了，形参传进来了）
     **/
    @Transactional
    public void createVoucherOrderBlock(VoucherOrder voucherOrder) {
        Long userId = voucherOrder.getUserId();
        int count = query().eq("user_id", userId).eq("voucher_id", voucherOrder.getVoucherId()).count();
        if (count > 0) {
            // 判断是否存在相关订单
            log.error("###### 用户已经使用此优惠券购买过一次");
            return;
        }

        // 扣减库存
        boolean flag = seckillVoucherService.update()
                .setSql("stock = stock-1")
                .eq("voucher_id", voucherOrder.getVoucherId())
                // 乐观锁 cas,库存大于0即可
                .gt("stock", 0)
                .update();
        if (!flag) {
            log.error("###### 扣减库存失败");
        }
        // 保存订单到 数据库
        save(voucherOrder);
    }

    // 通过类字段传递代理对象,用于获取代理对象事务
    private IVoucherOrderService proxy;

    /**
     * @param: [voucherId]
     * @return: com.hmdp.dto.Result
     * @description: 抢购优惠券，条件判断--redis的 Stream消息队列，一人一单，三代分布式锁
     **/
    @Override
    public Result seckillVoucher(Long voucherId) {
        Long userId = UserHolder.getUser().getId();
        long orderId = redisIDWorker.nextId("order");

        // 执行lua脚本,判断是否有库存  以及用户是否下过单
        // 扣除库存,添加用户id到订单set中(判断一人一单)
        // 发送消息(三个ARGV参数)到Stream消息队列
        Long result = stringRedisTemplate.execute(
                SECKILL_SCRIPT,
                Collections.emptyList(),
                // lua脚本ARGV参数列表
                voucherId,
                userId.toString(),
                String.valueOf(orderId));

        // 结果不为0不能下单
        int r = result.intValue();
        if (result == 0L) {
            return Result.fail(r == 1 ? "库存不足" : "同一用户不能重复下单");
        }

        // 获取代理对象
        proxy = (IVoucherOrderService) AopContext.currentProxy();

        // 返回订单id
        return Result.ok(orderId);
    }

    /*********************************************************************************************************/

    /**
     * @param: [voucherId]
     * @return: com.hmdp.dto.Result
     * @description: 抢购优惠券，条件判断--阻塞消息队列，一人一单，三代分布式锁
     **/
    /* @Override
    public Result seckillVoucher(Long voucherId) {
        Long userId = UserHolder.getUser().getId();
        // 执行lua脚本,判断是否有库存  以及用户是否下过单
        Long result = stringRedisTemplate.execute(
                SECKILL_SCRIPT,
                Collections.emptyList(),
                voucherId,
                userId.toString());

        // 结果不为0不能下单
        int r = result.intValue();
        if (result == 0L) {
            return Result.fail(r == 1 ? "库存不足" : "同一用户不能重复下单");
        }

        // 结果为0可以下单，下单信息存到 阻塞队列
        // 创建订单
        VoucherOrder voucherOrder = new VoucherOrder();
        // 订单id,生成一个全局id
        long orderId = redisIDWorker.nextId("order");
        voucherOrder.setId(orderId);
        // 用户id存在ThreadLocal里
        voucherOrder.setUserId(userId);
        // 代金券id
        voucherOrder.setVoucherId(voucherId);

        // 放入阻塞队列
        orderTasks.add(voucherOrder);


        // 获取代理对象
        proxy = (IVoucherOrderService) AopContext.currentProxy();

        return Result.ok(orderId);
    } */

    /**
     * @param: [voucherId]
     * @return: com.hmdp.dto.Result
     * @description: 抢购优惠券，条件判断，一人一单，三代分布式锁
     **/
    /* @Override
    public Result seckillVoucher(Long voucherId) {
        // 查询优惠券信息
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);

        LocalDateTime now = LocalDateTime.now();
        // 判断秒杀开始
        LocalDateTime beginTime = voucher.getBeginTime();
        if (beginTime.isAfter(now)) {
            // 开始时间晚于当前时间
            return Result.fail("秒杀尚未开始");
        }

        // 判断秒杀结束
        LocalDateTime endTime = voucher.getEndTime();
        if (endTime.isBefore(now)) {
            // 结束时间早于当前时间
            return Result.fail("秒杀已经结束");
        }

        // 库存是否充足
        Integer stock = voucher.getStock();
        if (stock < 1) {
            // 优惠券库存不足，无法下单
            return Result.fail("库存不足");
        }

        // 获取用户id
        UserDTO user = UserHolder.getUser();
        Long userId = user.getId();

         *//* // 悲观锁，保证某个用户一人一单查询订单数量的时候不会超买，使用 用户id 作为锁
        // 用悲观锁包裹整个事务,等待事务提交(整个函数执行完才提交)完毕后,再释放锁
        // 悲观锁不能应对分布式集群的情况
        synchronized (userId.toString().intern()) {
            // 获取代理对象（事务）
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId, userId);
        } *//*

     *//* // 自定义分布式锁
        SimpleRedisLock simpleRedisLock = new SimpleRedisLock("order:" + userId, new StringRedisTemplate());
        boolean flag = simpleRedisLock.tryLock(100);
        if (!flag) {
            // 获取锁失败，返回错误或者重试
            return Result.fail("同一用户不允许重复下单");
        }
        try {
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId, userId);
        } finally {
            // 释放锁
            simpleRedisLock.unLock();
        } *//*

        // Redisson分布式锁
        RLock lock = redissonClient.getLock("order:" + userId);
        boolean flag = lock.tryLock();
        if (!flag) {
            // 获取锁失败，返回错误或者重试
            return Result.fail("同一用户不允许重复下单");
        }
        try {
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId, userId);
        } finally {
            // 释放锁
            lock.unlock();
        }
    } */

    /** (无用了,过时的)
     * @param: [voucherId, userId]
     * @return: com.hmdp.dto.Result
     * @description: 扣减库存（乐观锁），创建订单（悲观锁确定是否存在相关订单）,外套一个悲观锁实现一人一单（insert）
     * 悲观锁不适用于集群环境
     **/
    @Transactional
    public Result createVoucherOrder(Long voucherId, Long userId) {
        // 一人一单
        // 查询订单数量
        int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
        if (count > 0) {
            // 判断是否存在相关订单
            return Result.fail("用户已经使用此优惠券购买过一次");
        }

        // 扣减库存
        boolean flag = seckillVoucherService.update()
                .setSql("stock = stock-1")
                .eq("voucher_id", voucherId)
                // 乐观锁 cas,库存大于0即可
                .gt("stock", 0)
                .update();
        if (!flag) {
            return Result.fail("扣减库存失败");
        }

        // 创建订单
        VoucherOrder voucherOrder = new VoucherOrder();
        // 订单id,生成一个全局id
        long orderId = redisIDWorker.nextId("order");
        voucherOrder.setId(orderId);
        // 用户id存在ThreadLocal里
        voucherOrder.setUserId(userId);
        // 代金券id
        voucherOrder.setVoucherId(voucherId);
        // 存储到数据库
        save(voucherOrder);

        // 返回创建的订单id
        return Result.ok(orderId);
    }
}
